package tjs.study.notes.dotest.disruptor.apiLog.config;

import com.alibaba.fastjson.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import tjs.study.notes.dotest.disruptor.apiLog.bean.ApiLogModelWriteDto;
import tjs.study.notes.dotest.disruptor.apiLog.service.WriteLogService;
import tjs.study.notes.dotest.disruptor.parallel.spring.DisruptorNotice;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * Created by Nicky.Tang on 2022/9/29 上午11:35
 *
 * @since 02.12.10
 */
@Lazy
@Scope(value = "prototype")
@Component
public class DisruptorAccessKeyConsumer implements IDisruptorConsume<DisruptorNotice> {
    private static final Logger logger = LoggerFactory.getLogger(DisruptorAccessKeyConsumer.class);
    @Autowired
    private WriteLogService writeLogService;
    @Autowired
    private DisruptorConfig disruptorConfig;
    private String consumerId;
    private Long lastTs = System.currentTimeMillis() + 20 * 1000;//最后一次写入es的时间戳；初始时间加上20秒，避免启动时isWrite马上就是true
    private List<ApiLogModelWriteDto> datas = new ArrayList<>(1024 * 2);

    public DisruptorAccessKeyConsumer(String consumerId) {
        this.consumerId = consumerId;
    }

    @Override
    public void onEvent(DisruptorNotice event) throws Exception {
        logger.trace("当前消费者: {}, 消费信息: {}", this.consumerId, event.getNoticeJson());
        datas.add(JSONObject.parseObject(event.getNoticeJson(), ApiLogModelWriteDto.class));
        this.tryConsume();
        //event.setNoticeJson(null);//todo-tjs 手动清空，避免oom。待验证
    }

    @Override
    public Boolean isWrite() {
        if (CollectionUtils.isEmpty(datas)) {
            return false;
        }
        logger.trace("size:{}?{}={},interval:{}？{}={}",
                datas.size(), disruptorConfig.getMaxPoolRecords(), datas.size() >= disruptorConfig.getMaxPoolRecords(),
                parseToStringTS(System.currentTimeMillis()), parseToStringTS(lastTs + disruptorConfig.getMaxPoolInterval()), System.currentTimeMillis() >= (lastTs + disruptorConfig.getMaxPoolInterval()));

        return datas.size() >= disruptorConfig.getMaxPoolRecords()
                || System.currentTimeMillis() >= (lastTs + disruptorConfig.getMaxPoolInterval());
    }

    /**
     * synchronized: 定时线程DisruptorCompensateThread是周期性执行，所以执行tryConsume时主要时间还是偏向锁，不影响性能
     */
    @Override
    public synchronized void tryConsume() {
        try {
            if (isWrite()) {
                writeLogService.doWriteLog(datas);
                datas.clear();
            }
        } finally {
            lastTs = System.currentTimeMillis();
        }
    }

    public static String parseToStringTS(Long value) {
        return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS").format(new Date(value));
    }
}